package com.ygx.flink.practice.operators;

import com.ygx.flink.practice.model.PvUv;
import com.ygx.flink.practice.model.UserBehavior;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DateFormat;
import java.text.SimpleDateFormat;

/**
 * @author YangGuoXiang
 * @version 1.0
 * @date 2022/3/30 11:50
 */
public class Accumulator extends KeyedProcessFunction<Long, UserBehavior, PvUv> {

    private Logger log = LoggerFactory.getLogger(this.getClass());
    private ValueState<Integer> pvState;
    private ValueState<RoaringBitmap> userIdState;

    @Override
    public void open(Configuration parameters) throws Exception {
        pvState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("pvState", Integer.class));
        userIdState = getRuntimeContext().getState(new ValueStateDescriptor<RoaringBitmap>("userIdState", RoaringBitmap.class));
    }

    @Override
    public void processElement(UserBehavior value, KeyedProcessFunction<Long, UserBehavior, PvUv>.Context ctx, Collector<PvUv> out) throws Exception {
        Integer pv = pvState.value();
        RoaringBitmap userIds = userIdState.value();
        if(userIds == null){
            pv = 0;
            userIds = new RoaringBitmap();
        }
        pv++;
        userIds.add(value.getUserId());
        PvUv pvuv = new PvUv();
        DateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:00");
        String hourStr = sdf.format(ctx.getCurrentKey());
        pvuv.setDt(hourStr);
        pvuv.setPv(pv);
        pvuv.setUv(userIds.getCardinality());
        log.info("dt:{} ; userId:{} ; pv:{} ; uv: {}", hourStr, value.getUserId(), pv, userIds.getCardinality());
        out.collect(pvuv);
        pvState.update(pv);
        userIdState.update(userIds);
    }
}
